package org.wenzi.com.function;


import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;

/**
 * @author zhaozuowen
 * @date 2021-10-25 14:09
 */
public class TopNFunction extends KeyedProcessFunction<Tuple, WindowResultFunction.ItemViewCount,String> {


    private int topSize;

    private ListState<WindowResultFunction.ItemViewCount> listState;

    public ListState<WindowResultFunction.ItemViewCount> getListState() {
        return listState;
    }

    public void setListState(ListState<WindowResultFunction.ItemViewCount> listState) {
        this.listState = listState;
    }

    public int getTopSize() {
        return topSize;
    }

    public void setTopSize(int topSize) {
        this.topSize = topSize;
    }

    public TopNFunction(int topSize) {
        this.topSize=topSize;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        ListStateDescriptor stateDescriptor = new ListStateDescriptor("itemState-state", WindowResultFunction.ItemViewCount.class);
         this.listState = getRuntimeContext().getListState(stateDescriptor);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        List<WindowResultFunction.ItemViewCount> list = new ArrayList<WindowResultFunction.ItemViewCount>();
        for (WindowResultFunction.ItemViewCount itemViewCount : this.listState.get()) {
            list.add(itemViewCount);
        }
        this.listState.clear();
        list.sort((o1, o2) -> (int) (o2.getViewCount()-o1.getViewCount()));
        // 将排名信息格式化成 String, 便于打印
        StringBuilder result = new StringBuilder();
        result.append("====================================\n");
        result.append("时间: ").append(new Timestamp(timestamp-1)).append("\n");

        for (int i=0;i<topSize;i++) {
            WindowResultFunction.ItemViewCount currentItem = list.get(i);
            // No1:  商品ID=12224  浏览量=2413
            result.append("No").append(i).append(":")
                    .append("  商品ID=").append(currentItem.itemId)
                    .append("  浏览量=").append(currentItem.viewCount)
                    .append("\n");
        }
        result.append("===========================" +
                "=========\n\n");

        out.collect(result.toString());
    }

    @Override
    public void processElement(WindowResultFunction.ItemViewCount itemViewCount, Context context, Collector<String> collector) throws Exception {
        this.listState.add(itemViewCount);
        context.timerService().registerEventTimeTimer(itemViewCount.getWindowEnd()+1);
    }
}
